import concurrent.futures
import decimal
import os
from pathlib import Path

import numpy as np
import pandas as pd
from dateutil.relativedelta import relativedelta
from loguru import logger
from retrying import retry
from tqdm import tqdm

from apps.pdd.pdd.clean.encrypt_utils import FileEncryptDecryptUtils
from components.config import WFQ_SOURCE_MYSQL_CONFIG
from utils.db.mysqldb import MysqlDB

RESULT_MALL_KEYS = [
    "version",
    "platform_id",
    "platform_name",
    "mall_id",
    "mall_name",
    "company_name_src",
    "taxpayer_id",
    "company_name",
    "month_sales",
    "month_sales_money",
    "total_sales",
    "total_sales_money",
    "percent_month_sales",
    "percent_month_sales_money",
    "percent_total_sales",
    "percent_total_sales_money",
    "goods_num",
    "self_support",
    "mall_url",
    "mall_open_time",
    "mall_open_time_year",
    "mall_status",
    "mall_end_time",
    "mall_platform_classification",
    "mall_business_classification",
    "mall_scope",
    "company_province",
    "company_city",
    "company_county",
    "company_town",
    "mall_location_code",
    "mall_send_province",
    "mall_send_city",
    "mall_send_county",
    "mall_send_location_code",
    "mall_desc",
    "mall_logo",
    "mall_star",
    "mall_label",
    "mall_rating_text",
    "mall_achievements_rank",
    "mall_review_num",
    "mall_positive_num",
    "mall_followers_num",
    "mall_service_score",
    "mall_details_label_list",
    "create_at",
    "update_at",
]

RESULT_GOODS_KEYS = [
    "version",
    "company_name",
    "mall_name",
    "mall_id",
    "goods_id",
    "goods_name",
    "price",
    "sale_num",
    "level_1",
    "level_2",
    "level_3",
    "goods_rank",
    "hd_url",
    "market_price",
    "normal_price",
    "tag_list",
    "update_at",
]
import datetime


class Export(FileEncryptDecryptUtils):
    def __init__(self):
        self._to_db = None
        self._df_old = None
        self._df_new = None
        self.datas_day = self._get_days(
            start_date=str((datetime.datetime.now().date() - relativedelta(months=1)).replace(day=25)),
            end_date=str(datetime.datetime.now().date().replace(day=10)),
        )
        self.datas_start = str((datetime.datetime.now().date() - relativedelta(months=1)).replace(day=25))
        self.datas_end = str((datetime.datetime.now().date()).replace(day=11))
        self.last_datas_day = self._get_days(
            start_date=str((datetime.datetime.now().date() - relativedelta(months=2)).replace(day=25)),
            end_date=str((datetime.datetime.now().date() - relativedelta(months=1)).replace(day=10)),
        )
        self.last_datas_start = str((datetime.datetime.now().date() - relativedelta(months=2)).replace(day=25))
        self.last_datas_end = str((datetime.datetime.now().date() - relativedelta(months=1)).replace(day=11))
        self.shop_info_sql = (
            """select version, platform_id, platform_name, mall_id, mall_name, company_name_src, taxpayer_id, company_name, month_sales, month_sales_money, total_sales, total_sales_money, percent_month_sales, percent_month_sales_money, percent_total_sales, percent_total_sales_money, goods_num, self_support, mall_url, mall_open_time, mall_open_time_year, mall_status, mall_end_time, mall_platform_classification, mall_business_classification, mall_scope, company_province, company_city, company_county, company_town, mall_location_code, mall_send_province, mall_send_city, mall_send_county, mall_send_location_code, mall_desc, mall_logo, mall_star, mall_label, mall_rating_text, mall_achievements_rank, mall_service_score, mall_details_label_list
        from result_pdd_shop_info where mall_id in ({}) """
            + f""" and version in ({self.datas_day})"""
        )
        self.good_info_sql = (
            """select  version, company_name, mall_name, mall_id, goods_id, goods_name, price, sale_num, level_1, level_2, level_3, goods_rank, hd_url, market_price, normal_price, tag_list
        from result_pdd_goods_detail where mall_id in ({}) """
            + f""" and version in ({self.datas_day})"""
        )

    def _get_days(self, start_date="2025-02-25", end_date="2025-03-10", to_list=False):
        res = []
        start_date = datetime.datetime.strptime(start_date, "%Y-%m-%d")
        end_date = datetime.datetime.strptime(end_date, "%Y-%m-%d")
        current_date = start_date
        while current_date <= end_date:
            date_str = current_date.strftime("%Y-%m-%d")
            current_date += datetime.timedelta(days=1)
            res.append(date_str)
        return res if to_list else ",".join([f"'{i}'" for i in res])

    @property
    def to_db(self):
        if self._to_db is None:
            self._to_db = MysqlDB(
                ip=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_IP"],
                port=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_PORT"],
                db=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_DB"],
                user_name=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_NAME"],
                user_pass=WFQ_SOURCE_MYSQL_CONFIG["MYSQL_USER_PASS"],
            )
        return self._to_db

    def get_mall_id_and_version(self):
        sql = f"""
        SELECT o.mall_id, o.version AS old_version, o.old_update_at, n.version AS new_version, n.new_update_at
        FROM (SELECT mall_id, version, update_at AS old_update_at
              FROM source_net_pdd_shop_info
              WHERE update_at BETWEEN '{self.last_datas_start}' AND '{self.last_datas_end}'
                AND (is_strict_all = 1 or flag = -1)) o
                 JOIN (SELECT mall_id, version, update_at AS new_update_at
                       FROM source_net_pdd_shop_info
                       WHERE update_at BETWEEN '{self.datas_start}' AND '{self.datas_end}'
                         AND (is_strict_all = 1 or flag = -1)) n ON o.mall_id = n.mall_id
        WHERE n.new_update_at BETWEEN DATE_ADD(o.old_update_at, INTERVAL 1 MONTH) - INTERVAL 5 DAY
                  AND DATE_ADD(o.old_update_at, INTERVAL 1 MONTH) + INTERVAL 5 DAY
        """
        print(sql)
        datas = self.to_db.find(sql, to_json=True)
        df = pd.DataFrame(datas)
        logger.info(f"查询到数据 {df.shape}")
        return df

    def get_query_params(self):

        df_mall = self.get_mall_id_and_version()
        df_no_dup_mall = df_mall.sort_values(["old_version", "new_version"], ascending=[True, False])
        df_no_dup_mall = df_no_dup_mall.drop_duplicates(subset=["mall_id"], keep="first")
        df_query = df_no_dup_mall[["mall_id", "old_version", "new_version"]]

        sqls = []
        for idx, i in df_query.iterrows():
            mall_id = i["mall_id"]
            version = tuple([i["old_version"], i["new_version"]])
            sqls.append(f" (mall_id = '{mall_id}' and version in {version}) ")

        return sqls

    def _find_datas(self, query):
        return pd.DataFrame(self.to_db.find(query, to_json=True))

    def get_mall_datas(self, sqls):
        sql = """select * from source_net_pdd_shop_info where {}"""
        df_all = pd.DataFrame()
        batch_size = 100

        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            tasks = [
                executor.submit(self._find_datas, sql.format(" or ".join(batch_sqls)))
                for batch_sqls in [sqls[i:i + batch_size] for i in range(0, len(sqls), batch_size)]
            ]
            for task in tqdm(concurrent.futures.as_completed(tasks), total=len(tasks), desc=f"任务 pid:{os.getpid()}"):
                if task.exception():
                    raise task.exception()
                df_all = pd.concat([df_all, task.result()])

        return df_all

    def get_goods_datas(self, sqls):
        sql = """select * from source_result_pdd_goods_detail where {}"""
        df_all = pd.DataFrame()
        batch_size = 100

        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
            tasks = [
                executor.submit(self._find_datas, sql.format(" or ".join(batch_sqls)))
                for batch_sqls in [sqls[i:i + batch_size] for i in range(0, len(sqls), batch_size)]
            ]
            for task in tqdm(concurrent.futures.as_completed(tasks), total=len(tasks), desc=f"任务 pid:{os.getpid()}"):
                if task.exception():
                    raise task.exception()
                df_all = pd.concat([df_all, task.result()])

        return df_all

    @logger.catch
    def process_mall_data(self, df_pro: pd.DataFrame):
        new_df = df_pro.copy()
        new_df.sort_values(by=["version"], ascending=[False])
        new_df.rename(
            columns={
                "shop_name": "mall_name",
                "shop_url": "mall_url",
                "shop_status": "mall_status",
                "service_score": "mall_service_score",
                "calculate_sales": "total_sales",
                "crawler_goods_num": "goods_num",
                "calculate_sales_money": "total_sales_money",
            },
            inplace=True,
        )
        new_df["total_sales"].fillna(decimal.Decimal(0), inplace=True)
        new_df["total_sales_money"].fillna(decimal.Decimal(0), inplace=True)
        new_df["total_sales"] = new_df["total_sales"] / 10000
        new_df["total_sales_money"] = new_df["total_sales_money"] / 10000
        new_df["mall_open_time_year"] = new_df["open_time_label"].map(lambda x: x.split("年")[0] if x else None)
        new_df["mall_label"] = new_df["mall_label_volist"].map(
            lambda x: ";".join([i["desc"] for i in x]) if x else None
        )
        new_df["mall_achievements_rank"] = new_df["mall_label_volist"].map(
            lambda x: ";".join([i["desc"] for i in x if "榜第" in i["desc"]]) if x else None
        )
        new_df["mall_rating_text"] = new_df["mall_rating_text_list"].map(
            lambda x: (
                ";".join(
                    [
                        (
                            (i["mallRatingKey"]["txt"] + ":" + i["mallRatingValue"]["txt"])
                            if "mallRatingKey" in i
                            else (i["mall_rating_key"]["txt"] + ":" + i["mall_rating_value"]["txt"])
                        )
                        for i in x
                    ]
                )
                if x
                else None
            )
        )

        min_version = df_pro["version"].min()
        max_version = df_pro["version"].max()

        old_sales = new_df[new_df["version"] == min_version]["total_sales"].tolist()[0]
        new_sales = new_df[new_df["version"] == max_version]["total_sales"].tolist()[0]

        old_sales_money = new_df[new_df["version"] == min_version]["total_sales_money"].tolist()[0]
        new_sales_money = new_df[new_df["version"] == max_version]["total_sales_money"].tolist()[0]

        old_sales = decimal.Decimal(old_sales)
        new_sales = decimal.Decimal(new_sales)
        old_sales_money = decimal.Decimal(old_sales_money)
        new_sales_money = decimal.Decimal(new_sales_money)

        result_df = new_df[[i for i in RESULT_MALL_KEYS if i in new_df.columns]]
        res = []
        result_df.replace({np.nan: None}, inplace=True)
        rows = result_df.to_dict("records")
        for row in rows:
            if row["version"] == min_version:
                row["month_sales"] = old_sales
                row["month_sales_money"] = old_sales_money
                row["percent_month_sales"] = 100
                row["percent_month_sales_money"] = 100
                row["percent_total_sales"] = 100
                row["percent_total_sales_money"] = 100
            else:
                if row["mall_status"] in ["该店铺正在上传商品"]:
                    row["total_sales"] = None
                    row["total_sales_money"] = None
                    row["month_sales"] = None
                    row["month_sales_money"] = None
                    row["percent_month_sales"] = None
                    row["percent_month_sales_money"] = None
                    row["percent_total_sales"] = None
                    row["percent_total_sales_money"] = None
                else:
                    row["month_sales"] = new_sales - old_sales
                    row["month_sales_money"] = new_sales_money - old_sales_money
                    row["percent_month_sales"] = (
                        None if old_sales == 0 else round((new_sales - old_sales) / old_sales * 100, 4)
                    )
                    row["percent_month_sales_money"] = (
                        None
                        if old_sales_money == 0
                        else round((new_sales_money - old_sales_money) / old_sales_money * 100, 4)
                    )
                    row["percent_total_sales"] = (
                        None if old_sales == 0 else round((new_sales - old_sales) / old_sales * 100, 4)
                    )
                    row["percent_total_sales_money"] = (
                        None
                        if old_sales_money == 0
                        else round((new_sales_money - old_sales_money) / old_sales_money * 100, 4)
                    )
            res.append(row)
        return pd.DataFrame(res)

    def process_mall_base_info(self, sqls):
        df_all = self.get_mall_datas(sqls)
        mall_info_df = df_all.groupby("mall_id").apply(lambda x: self.process_mall_data(x))
        mall_info_df.replace({np.nan: None}, inplace=True)
        mall_info_rows = mall_info_df.to_dict("records")
        batch_size = 1000
        for batch_rows in [mall_info_rows[i:i + batch_size] for i in range(0, len(mall_info_rows), batch_size)]:
            self.to_db.add_batch_smart("result_pdd_shop_info", batch_rows, update_columns=list(batch_rows[0].keys()))

    def process_goods_data(self, df_pro: pd.DataFrame):
        return df_pro

    def process_goods_info(self, sqls):
        goods_df = self.get_goods_datas(sqls)
        goods_df = goods_df.drop(columns=["sale_num"])
        goods_df.rename(columns={"sale_num_by_tips": "sale_num"}, inplace=True)
        goods_df.replace({np.nan: None}, inplace=True)
        goods_df = goods_df[[i for i in RESULT_GOODS_KEYS if i in goods_df.columns]]
        goods_df_rows = goods_df.to_dict("records")
        batch_size = 1000
        for batch_rows in [goods_df_rows[i:i + batch_size] for i in range(0, len(goods_df_rows), batch_size)]:
            self.to_db.add_batch_smart("result_pdd_goods_detail", batch_rows, update_columns=list(batch_rows[0].keys()))

    @retry(stop_max_attempt_number=5)
    @logger.catch
    def get_datas_from_db(self, sql):
        return self.to_db.find(sql, to_json=True)

    def get_mall_info(self, mall_ids, batch_size=1000):
        datas = []
        for idx, i in enumerate(range(0, len(mall_ids), batch_size)):
            batch_mall_ids_str = ",".join([f"'{i}'" for i in mall_ids[i:i + batch_size]])
            batch_data = self.get_datas_from_db(self.shop_info_sql.format(batch_mall_ids_str))
            datas.extend(batch_data)

        return pd.DataFrame(datas)

    def get_goods_info(self, mall_ids, batch_size=1000):
        datas = []
        for idx, i in enumerate(range(0, len(mall_ids), batch_size)):
            mall_ids_str = ",".join([f"'{i}'" for i in mall_ids[i:i + batch_size]])
            data = self.get_datas_from_db(self.good_info_sql.format(mall_ids_str))
            if data:
                datas.extend(data)
        return pd.DataFrame(datas)

    @logger.catch
    def run_by_ids(self, mall_ids):
        df_mall = self.get_mall_info(mall_ids)
        df_goods = self.get_goods_info(mall_ids)
        return df_mall, df_goods

    def run_mul_by_ids(self, all_mall_ids, max_workers=10, batch_size=10000):
        df_malls = pd.DataFrame()
        df_goods = pd.DataFrame()
        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_result = {}
            for i in range(0, len(all_mall_ids), batch_size):
                mall_ids = all_mall_ids[i:i + batch_size]
                future_to_result.update({executor.submit(self.run_by_ids, mall_ids): mall_ids})

            for future in tqdm(
                concurrent.futures.as_completed(future_to_result),
                total=len(future_to_result),
            ):
                mall_ids = future_to_result[future]
                try:
                    df_mall, df_good = future.result()
                    df_malls = pd.concat([df_malls, df_mall], axis=0)
                    df_goods = pd.concat([df_goods, df_good], axis=0)
                except Exception as e:
                    logger.error(f"{mall_ids} 错误: {e}")

        path = Path(__file__).parent / datetime.datetime.now().strftime("%Y%m%d")
        path.mkdir(parents=True, exist_ok=True)
        df_malls.reset_index(drop=True)
        df_goods.reset_index(drop=True)
        df_malls.sort_values(by=["goods_num", "mall_open_time_year"], ascending=False, inplace=True)
        df_malls.drop_duplicates(subset=["mall_id"], inplace=True, keep="first")
        df_goods.drop_duplicates(subset=["mall_id", "goods_id"], inplace=True)
        print("生成结果：", df_malls.shape)

        # 每 batch_size 生成一个文件
        # 多线程生成

        with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
            future_to_result = {}

            for idx, i in enumerate(range(0, len(df_malls), batch_size), start=1):
                batch_mall = df_malls[i:i + batch_size]
                batch_good = df_goods[df_goods["mall_id"].isin(df_malls["mall_id"])]
                logger.info(f"{idx} 开始")
                future_to_result.update({executor.submit(self.gen_csv, batch_mall, batch_good, idx): idx})

            for future in tqdm(
                concurrent.futures.as_completed(future_to_result),
                total=len(future_to_result),
            ):
                idx = future_to_result[future]
                try:
                    future.result()
                except Exception as e:
                    logger.error(f"{idx} 错误: {e}")
                finally:
                    logger.info(f"{idx} 完成")

        self.encrypt_file()

    def gen_csv(self, df_malls, df_goods, idx):
        time_str = datetime.datetime.now().strftime("%Y%m%d")
        path = Path(__file__).parent / time_str
        path.mkdir(parents=True, exist_ok=True)
        encrypt_path = Path(__file__).parent / time_str / "encrypt"
        encrypt_path.mkdir(parents=True, exist_ok=True)

        mall_file_name = f"repay_plan_mall_{time_str}_{str(idx).zfill(4)}0000.csv"
        df_malls.to_csv(path / mall_file_name, index=False)
        df_malls = df_malls.reset_index(drop=True)
        this_goods_df = df_goods.reset_index(drop=True)

        if this_goods_df.shape[0] > 1048570:
            # 超过最大行数，分批写入
            for idx_j, j in enumerate(range(0, this_goods_df.shape[0], 1048570), start=1):
                goods_file_name = f"repay_plan_goods_{time_str}_{str(idx).zfill(4)+str(idx_j).zfill(4)}.csv"
                goods_res = this_goods_df[j:j + 1048570].reset_index(drop=True)
                if not goods_res.empty:
                    goods_res.to_csv(path / goods_file_name, index=False)
        else:
            goods_file_name = f"repay_plan_goods_{time_str}_{str(idx).zfill(4)}0000.csv"
            goods_res = df_goods[df_goods["mall_id"].isin(df_malls["mall_id"])].reset_index(drop=True)
            if not goods_res.empty:
                goods_res.to_csv(path / goods_file_name, index=False)

    def encrypt_file(self):
        time_str = datetime.datetime.now().strftime("%Y%m%d")
        path = Path(__file__).parent / time_str
        path.mkdir(parents=True, exist_ok=True)
        encrypt_path = Path(__file__).parent / time_str / "encrypt"
        encrypt_path.mkdir(parents=True, exist_ok=True)

        print("开始加密")
        for p in path.iterdir():
            if "repay_plan" in p.as_posix():
                self.encrypt(p, encrypt_path / p.name)
        print("加密完成")

    def get_now_ids(self):
        sql = f"""
        SELECT mall_id
        FROM result_pdd_shop_info
        WHERE version in ({self.datas_day})
        """
        # sql += """ and month_sales is not null """
        print(sql)
        datas = self.to_db.find(sql, to_json=True)
        df = pd.DataFrame(data=datas)
        return df["mall_id"].tolist()

    def run(self):
        sqls = self.get_query_params()
        logger.info("开始查询")
        logger.info("处理店铺")
        self.process_mall_base_info(sqls)
        logger.info("处理商品")
        self.process_goods_info(sqls)
        self.export()

    def export(self):
        logger.info("开始导出")
        ids = self.get_now_ids()
        self.run_mul_by_ids(ids)


if __name__ == "__main__":
    # Export().run()
    Export().get_mall_id_and_version()
